package miaodongsql;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 实时统计当前每个等级的用户数
 */
public class D1_stats_level {

  public static void main(String[] args) {

    Configuration configuration = new Configuration();
    configuration.setString("rest.port", "9091");
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
    env.setParallelism(1);
    StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


    String genSql = "CREATE TABLE ods_tb ( " +
        " stime STRING," +
        " name STRING," +
        " level BIGINT," +
        " rowtime AS to_timestamp(stime)," +
        " WATERMARK FOR rowtime AS rowtime - interval '30' second" +
        ") WITH ( " +
        "  'connector' = 'kafka'," +
        "  'topic' = 'test'," +
        "  'properties.bootstrap.servers' = 'wsl:9092'," +
        "  'properties.group.id' = 'testGroup'," +
        "  'scan.startup.mode' = 'latest-offset'," +
        "  'format' = 'json'" +
        ")";


    String pprint = "CREATE TABLE pprint (" +"    " +
        "    stime timestamp, " +
        "    level BIGINT, " +
        "    cnt BIGINT " +
        ") WITH (" +
        "    'connector' = 'print'" +
        ")";

    /**
     * 每个用户应该只取最新的等级，使用row_number取每个用户的最新等级
     * 这里的输出数据会包含回撤
     */
    String sql = "CREATE VIEW v_base AS" +
        " SELECT name,level FROM " +
        " (SELECT *,row_number()OVER(PARTITION BY name order by rowtime desc) as rownum from ods_tb) a where rownum=1";

    String insertSql = "INSERT INTO pprint " +
        " SELECT CURRENT_TIMESTAMP, level,count(1) as cnt from v_base GROUP BY level";


    tableEnv.executeSql(genSql);
    tableEnv.executeSql(pprint);
    tableEnv.executeSql(sql);
    tableEnv.executeSql(insertSql);

  }

}
